// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: chensili

#include <cmath>
#include <memory>
#include <string>
#include <utility>

#include "coin2/strategy/util/health_reporter_util.h"
#include "coin2/strategy/util/latency_util.h"
#include "coin2/strategy/dmm_strat/dmm_strategy.h"

namespace coin2::strategy::dmm_strat {

coin::proto::LatencyProto::LatencyTag GetFeedLatencyTag(const FeedUpdate& upd) {
  if (upd.is_book())
    return coin::proto::LatencyProto::BOOK_FEED_LATENCY;
  else if (upd.is_trade())
    return coin::proto::LatencyProto::TRADE_FEED_LATENCY;
  else
    return coin::proto::LatencyProto::FEED_LATENCY;
}

coin::proto::AccountRequestProto GetAccRequest(const MarketExchangeApi& mea) {
  coin::proto::AccountRequestProto acc_request;
  acc_request.set_market_type(MarketType_Name(mea.market));
  acc_request.set_exchange(ExchangeType_Name(mea.exchange));
  acc_request.set_api_version(mea.api);
  return acc_request;
}

std::string GetProductKey(const IProduct* product) {
  return fmt::format("{}.{}", product->absolute_norm(), product->exchange_name());
}

std::string GetProductKey(const OrderEvent* oe) {
  return fmt::format("{}.{}", oe->symbol(), ExchangeTypeToString(oe->exchange_type()));
}

bool HasMidPrice(const FeedUpdate& upd) {
  bool has_midp = false;
  if (upd.book().Bid0().has_value() &&
      upd.book().Ask0().has_value() &&
      upd.book().Bid0().value().price < upd.book().Ask0().value().price) {
    has_midp = true;
  }
  return has_midp;
}

std::optional<double> GetMidPrice(const FeedUpdate& upd) {
  if (HasMidPrice(upd)) {
    return (upd.book().Bid0().value().price + upd.book().Ask0().value().price) / 2;
  } else {
    return std::nullopt;
  }
}

double AdjustPriceByBound(double price, double upper_bound, double lower_bound) {
  CHECK(upper_bound > lower_bound) << upper_bound << " vs " << lower_bound;
  auto adj_price = std::min(price, upper_bound);
  adj_price = std::max(price, lower_bound);
  return adj_price;
}

bool WithinRolloverPeriod(const IProduct* product, int64_t timestamp) {
  bool ret = false;
  auto pt = PtimeFromTimestamp(timestamp);
  if (product->exchange() == ExchangeType::Okex ||
      product->exchange() == ExchangeType::Huobi) {
    if (product->market() == MarketType::Futures) {
      if (pt.date().day_of_week() == boost::date_time::Friday &&
          pt.time_of_day().hours() == 7 && pt.time_of_day().minutes() >= 55 &&
          pt.time_of_day().hours() == 8 && pt.time_of_day().minutes() <= 5) {
        ret = true;
      }
    }
  }
  return ret;
}

void ConvertExecutorConfig(coin2::strategy::PassiveExecutorConfig* executor_config) {
  if (executor_config->has_order_update_period_sec()) {
    executor_config->set_order_update_period(
        executor_config->order_update_period_sec() * 1e9);
  }
  if (executor_config->has_min_posting_period_sec()) {
    executor_config->set_min_posting_period(
        executor_config->min_posting_period_sec() * 1e9);
  }
  if (executor_config->has_max_posting_period_sec()) {
    executor_config->set_max_posting_period(
        executor_config->max_posting_period_sec() * 1e9);
  }
  if (executor_config->has_fill_cooldown_period_sec()) {
    executor_config->set_fill_cooldown_period(
        executor_config->fill_cooldown_period_sec() * 1e9);
  }
}

void InitCoin1ExecutorConfig(
    const coin2::strategy::PassiveExecutorConfig& coin2_exe_config,
    AggressiveExecutorConfig* coin1_exe_config) {
  CHECK(coin2_exe_config.has_symbol());
  CHECK(coin2_exe_config.has_mea());
  CHECK(coin2_exe_config.has_lot_size());
  CHECK(coin2_exe_config.has_min_pos());
  CHECK(coin2_exe_config.has_max_pos());
  auto mea = MarketExchangeApi::FromString(coin2_exe_config.mea());
  auto product = CreateProductFromNormString(
      mea, coin2_exe_config.symbol(), GetCurrentTimestamp());
  coin1_exe_config->set_symbol(product->symbol());
  coin1_exe_config->set_lot_size(coin2_exe_config.lot_size());
  coin1_exe_config->set_min_pos(coin2_exe_config.min_pos());
  coin1_exe_config->set_max_pos(coin2_exe_config.max_pos());
}

AmplitudeChecker::AmplitudeChecker(
    const IProduct* product, const AmplitudeCheckerConfig& config)
  : product_key_(GetProductKey(product)),
    config_(config) {
  CHECK(config_.has_window_min());
  CHECK(config_.has_threshold_bp());
  moving_low_.reset(new MovingLow<double>(config.window_min() * 60e9));
  moving_high_.reset(new MovingHigh<double>(config.window_min() * 60e9));
  threshold_ = config_.threshold_bp() / 1e4;
  CHECK_GT(threshold_, 0);
}

void AmplitudeChecker::SetAlarm() {
  if (!alarm_) {
    alarm_ = true;
    LOG(INFO) << "Amplitude alarm set. " << product_key_;
  }
}

void AmplitudeChecker::ResetAlarm() {
  if (alarm_) {
    alarm_ = false;
    LOG(INFO) << "Amplitude alarm reset. " << product_key_;
  }
}

void AmplitudeChecker::onBookFeed(const std::string& product_key, const FeedUpdate& upd) {
  CHECK(product_key == product_key_) << product_key << " vs " << product_key_;
  auto midp = GetMidPrice(upd);
  if (midp.has_value()) {
    auto book_ts = upd.timestamp();
    moving_low_->Add(book_ts, midp.value());
    moving_high_->Add(book_ts, midp.value());
  }
}

void AmplitudeChecker::UpdateAlarm() {
  if (!moving_low_->Empty() && !moving_high_->Empty()) {
    double min_price = moving_low_->Get();
    double max_price = moving_high_->Get();
    double amplitude = max_price / min_price - 1;
    LOG(INFO) << "min_price: " << min_price << " "
              << "max_price: " << max_price << " "
              << "amplitude: " << amplitude;
    if (amplitude >= threshold_) {
      SetAlarm();
    } else if (amplitude <= 0.5 * threshold_) {
      ResetAlarm();
    }
  }
}

DmmPricer::DmmPricer(const IProduct* product, ::coin2::app::Driver* driver) {
  get_product_holder_ = [product, driver](int64_t ts) {
    return driver->product_holder_manager()->GetProductHolder(*product, ts);
  };
  product_key_ = GetProductKey(product);
}

void DmmPricer::onBookFeed(const std::string& product_key, const FeedUpdate& upd) {
  CHECK(product_key == product_key_) << product_key << " vs " << product_key_;
  if (HasMidPrice(upd)) {
    bid0_ = upd.book().Bid0();
    ask0_ = upd.book().Ask0();
    book_ts_ = upd.book().Timestamp();
    ready_ = true;
  } else {
    ready_ = false;
    LOG(INFO) << fmt::format(
        "Abnormal feed. product: {}", upd.product().absolute_norm());
  }
}

std::pair<double, double> DmmPricer::CalculatePriceBySpread(
    double spread, bool bbo_protection) {
  CHECK(Ready());
  double midp = (bid0_.value().price + ask0_.value().price) / 2;
  // 0.1bp for buffer
  double askp = midp * (1 + spread / 2 - 0.00001);
  double bidp = midp * (1 - spread / 2 + 0.00001);
  double cal_spread = (askp - bidp) / midp;
  CHECK(cal_spread <= spread) << cal_spread << "vs" << spread;
  return AdjustPrice(askp, bidp, bbo_protection);
}

std::pair<double, double> DmmPricer::CalculatePriceByEdge(
    double sell_edge, double buy_edge, bool bbo_protection) {
  CHECK(Ready());
  double askp = ask0_.value().price * (1 + sell_edge);
  double bidp = bid0_.value().price * (1 - buy_edge);
  return AdjustPrice(askp, bidp, bbo_protection);
}

std::pair<double, double> DmmPricer::AdjustPrice(
      double sell_price, double buy_price, bool bbo_protection) {
  CHECK(Ready());
  const auto& tick_price_policy = get_product_holder_(*book_ts_)->product_info().tick_price();
  double askp = tick_price_policy.RoundAggressively(sell_price, OrderDirection::DIRECTION_SELL);
  double bidp = tick_price_policy.RoundAggressively(buy_price, OrderDirection::DIRECTION_BUY);
  double ask0 = ask0_.value().price;
  double bid0 = bid0_.value().price;
  double ask0_minus_one = tick_price_policy.GetPrevPrice(ask0);
  double bid0_plus_one = tick_price_policy.GetNextPrice(bid0);
  if (bbo_protection) {
    askp = std::max({askp, ask0, bid0_plus_one});
    bidp = std::min({bidp, ask0_minus_one, bid0});
  } else {
    askp = std::max({askp, ask0_minus_one, bid0_plus_one});
    bidp = std::min({bidp, ask0_minus_one, bid0_plus_one});
  }
  return {askp, bidp};
}

DmmStrategy::DmmStrategy(const std::experimental::filesystem::path& strat_json,
                         bool dry_run,
                         std::optional<int64_t> timestamp,
                         int exit_after_sec )
  : strategy_created_time_(timestamp.has_value() ? *timestamp : GetCurrentTimestamp()),
    exit_after_ns_(exit_after_sec * 1e9) {
  std::ifstream in(strat_json);
  LOG(INFO) << "Load strategy json: " << strat_json;
  CHECK_THROW(in.is_open());
  auto strat_config_str = ::nlohmann::json::parse(in).dump();
  DmmStrategyConfig dmm_strat_config;
  auto status = google::protobuf::util::JsonStringToMessage(
      strat_config_str, &dmm_strat_config);
  CHECK_THROW(status.ok()) << "Fail to parse DriverConfig from " << strat_config_str
                           << "\n" << status;
  for (auto& product_config : *dmm_strat_config.mutable_product_configs()) {
    product_config.mutable_pass_config()->set_symbol(product_config.trade_symbol());
    product_config.mutable_pass_config()->set_mea(product_config.trade_mea());
    ConvertExecutorConfig(product_config.mutable_pass_config());
  }
  LOG(INFO) << dmm_strat_config.DebugString();
  for (auto& product_config : dmm_strat_config.product_configs()) {
    dmm_sub_strats_.push_back(std::make_unique<DmmSubStrategy>(
        this,
        product_config,
        strategy_created_time_,
        dmm_strat_config.enable_telemetry(),
        dry_run));
  }
  for (auto& dmm_sub_strat : dmm_sub_strats_) {
    const auto& key = GetProductKey(dmm_sub_strat->GetTradeProduct());
    feed_sub_map_[key].push_back(dmm_sub_strat.get());
    CHECK(order_sub_map_.find(key) == order_sub_map_.end());
    order_sub_map_[key] = dmm_sub_strat.get();
  }
}

void DmmStrategy::InitAuthKey(::coin2::app::Driver* driver) {
  const auto& driver_config = driver->config();
  for (const auto& [exchange, order_config] : driver_config.order().exchanges()) {
    CHECK(!auth_key_);
    const auto& key_filepath = order_config.connection_config().key_filepath();
    auth_key_.reset(new AuthKey(AuthKey::FromFile(key_filepath)));
  }
}

void DmmStrategy::Init(::coin2::app::Driver* driver) {
  LOG(INFO) << "Dmm Strategy Init.";
  InitAuthKey(driver);
  strat_reporter_.reset(new StrategyReporter2(
      driver->strategy()->config()));
  for (auto& sub_strat : dmm_sub_strats_) {
    sub_strat->Init(driver, strat_reporter_.get());
    auto acc_request = GetAccRequest(sub_strat->mea());
    acc_request.set_owner(auth_key_->owner());
    auto curr_time = GetCurrentTimestamp();
    strat_reporter_->RegisterExecutorConfig(
        curr_time, acc_request,
        *sub_strat->GetTradeProduct(), sub_strat->GetConfig().pass_config());
  }
}

void DmmStrategy::InjectFeed(const coin2::feed::mgr::impl::FeedSubsystemManager* feed_mgr) {
  for (auto& elem : feed_sub_map_) {
    for (auto& sub_strat : elem.second) {
      sub_strat->InjectFeed(feed_mgr);
    }
  }
}

void DmmStrategy::onBookFeed(const FeedUpdate& upd) {
  // DLOG(INFO) << "Feed received.";
  if (GetCurrentTimestamp() - strategy_created_time_ > exit_after_ns_) {
    std::exit(0);
  }
  const auto& key = GetProductKey(&upd.product());
  CHECK_GT(feed_sub_map_[key].size(), 0);
  for (auto& sub_strat : feed_sub_map_[key]) {
    sub_strat->onBookFeed(key, upd);
  }
  if (order_sub_map_.find(key) != order_sub_map_.end()) {
    auto midp = GetMidPrice(upd);
    if (midp.has_value()) {
      strat_reporter_->UpdateMidp(*order_sub_map_[key]->GetTradeProduct(), midp);
    }
  }
}

void DmmStrategy::onAccountInfo(const OrderUpdate& upd) {
  for (auto& sub_strat : dmm_sub_strats_) {
    sub_strat->onAccountInfo(upd);
  }
  if (strat_reporter_) {
    strat_reporter_->onAccountInfo(upd);
  }
}

void DmmStrategy::onAccountOrder(const OrderUpdate& upd) {
  if (upd.product_order_info().event().type() == OrderEvent::ORDER_FILLED) {
    const auto& key = GetProductKey(&upd.product_order_info().event());
    if (order_sub_map_.count(key) == 0) {
      LOG(INFO) << fmt::format("Received unexpected fills {}", key);
    } else {
      order_sub_map_[key]->onAccountOrder(upd);
    }
  }
}

void DmmStrategy::onOrderLog(const coin::proto::OrderGatewayLog& order_log) {
  if (strat_reporter_) {
    strat_reporter_->GetMutableStrategyLogger()->WriteOrderLog(
        order_log,
        true,
        GetCurrentTimestamp());
  }
}

void DmmStrategy::onCleanup(Driver* driver) {
  for (auto& sub_strat : dmm_sub_strats_) {
    auto executor = sub_strat->GetMutableExecutor();
    if (executor) {
      executor->ForceCancelAllWorkingOrders();
    }
  }
}

DmmSubStrategy::DmmSubStrategy(DmmStrategy* parent,
                               const DmmProductConfig& product_config,
                               int64_t strategy_created_time,
                               bool enable_telemetry,
                               bool dry_run)
  : parent_(parent),
    config_(product_config),
    strategy_created_time_(strategy_created_time),
    driver_(nullptr),
    strat_logger_(nullptr),
    health_reporter_(nullptr),
    enable_telemetry_(enable_telemetry),
    dry_run_(dry_run),
    bbo_protection_(false) {
  CHECK(parent_);
  CHECK(config_.has_trade_symbol());
  CHECK(config_.has_trade_mea());
  CHECK(config_.has_pass_config());
  trade_product_ = CreateProductFromNormString(
      MarketExchangeApi::FromString(config_.trade_mea()),
      config_.trade_symbol(),
      strategy_created_time);
  if (product_config.has_spread_bp()) {
    spread_ = product_config.spread_bp() / 1e4;
  }
  if (product_config.has_spread_bp_hard()) {
    spread_hard_ = product_config.spread_bp_hard() / 1e4;
  }
  if (product_config.has_bbo_protection()) {
    bbo_protection_ = product_config.bbo_protection();
  }
  if (product_config.has_reduce_only_with_position_threshold()) {
    reduce_only_with_position_threshold_ =
        product_config.reduce_only_with_position_threshold();
  }
  if (product_config.has_amplitude_checker_config()) {
    amplitude_checker_.reset(new AmplitudeChecker(
        GetTradeProduct(),
        product_config.amplitude_checker_config()));
  }
  InitCoin1ExecutorConfig(config_.pass_config(), &exe_config_);
  feed_keys_.insert(GetProductKey(GetTradeProduct()));
}

void DmmSubStrategy::Init(
    ::coin2::app::Driver* driver,
    ::coin2::strategy::util::StrategyReporter2* reporter) {
  driver_ = driver;
  strat_logger_ = reporter->GetMutableStrategyLogger();
  health_reporter_ = reporter->GetMutableHealthReporter();
  auto get_product_holder = [this](const coin2::exchange::base::symbology::IProduct& product, int64_t timestamp) {
      return driver_->product_holder_manager()->GetProductHolder(product, timestamp);
  };

  pricer_ = std::make_unique<DmmPricer>(trade_product_.get(), driver);
  executor_ = std::make_unique<PassiveOrderExecutor>(
      config_.pass_config(), os(), strategy_created_time_, nullptr, get_product_holder);
  if (enable_telemetry_) {
    LOG(INFO) << "Set Latency Recorder";
    os()->gateway()->set_latency_recorder(
         strat_logger_->mutable_latency_recorder());
  }
}

void DmmSubStrategy::InjectFeed(
    const coin2::feed::mgr::impl::FeedSubsystemManager* feed_mgr) {
  feed_mgr_ = feed_mgr;
}

void DmmSubStrategy::CheckRolloverPeriod(int64_t timestamp) {
  bool within_rollover_period = WithinRolloverPeriod(
      GetTradeProduct(), timestamp);
  if (within_rollover_period) {
    LOG(INFO) << "Within rollover period.";
  }
  if (within_rollover_period_ && within_rollover_period) {
    LOG(INFO) << "Exit after rollover to update product.";
    parent_->onCleanup(driver_);
    std::exit(0);
  }
  within_rollover_period_ = within_rollover_period;
}

bool DmmSubStrategy::ReduceOnlyEnforced() {
  bool amplitude_high = false;
  if (amplitude_checker_.get()) {
    amplitude_high = amplitude_checker_->GetAlarm();
  }
  bool reduce_only_with_position = false;
  if (reduce_only_with_position_threshold_.has_value()) {
    double reserve = 0.5 * (config_.pass_config().min_pos() +
                            config_.pass_config().max_pos());
    double pos = executor_->GetPosition() - reserve;
    if (std::abs(pos) > reduce_only_with_position_threshold_.value()) {
      reduce_only_with_position = true;
    }
  }
  bool reduce_only = false;
  if (within_rollover_period_ || amplitude_high || reduce_only_with_position) {
    reduce_only = true;
  }
  return reduce_only;
}

void DmmSubStrategy::onAccountInfo(const OrderUpdate& upd) {
  if (GetMutableExecutor() && pricer_->Ready()) {
    executor_->onAccountInfo(upd);
  }
}

void DmmSubStrategy::onBookFeed(const std::string& product_key, const FeedUpdate& upd) {
  auto curr_ts = upd.timestamp();
  CHECK(feed_keys_.find(product_key) != feed_keys_.end());
  if (last_rollover_check_ts_ + 10'000'000'000LL < curr_ts) {
    CheckRolloverPeriod(curr_ts);
    last_rollover_check_ts_ = curr_ts;
  }
  if (amplitude_checker_.get()) {
    amplitude_checker_->onBookFeed(product_key, upd);
  }
  if (last_amplitude_check_ts_ + 10'000'000'000LL < curr_ts) {
    amplitude_checker_->UpdateAlarm();
    last_amplitude_check_ts_ = curr_ts;
  }
  pricer_->onBookFeed(product_key, upd);
  if (os()->is_ready() && pricer_->Ready() && GetMutableExecutor()) {
    auto prices = pricer_->CalculatePriceBySpread(spread_.value(), bbo_protection_);
    auto askp = prices.first;
    auto bidp = prices.second;
    bool reduce_only = ReduceOnlyEnforced();
    executor_->SetReduceOnly(reduce_only);
    if (reduce_only) {
      askp = upd.book().Ask0().value().price;
      bidp = upd.book().Bid0().value().price;
    }
    if (spread_hard_.has_value()) {
      prices = pricer_->CalculatePriceBySpread(spread_hard_.value(), bbo_protection_);
      auto upper_bound = prices.first;
      auto lower_bound = prices.second;
      askp = AdjustPriceByBound(askp, upper_bound, lower_bound);
      bidp = AdjustPriceByBound(bidp, upper_bound, lower_bound);
    }
    prices = pricer_->AdjustPrice(askp, bidp, bbo_protection_);
    askp = prices.first;
    bidp = prices.second;
    auto after_feature_calc_time = GetCurrentTimestamp();
    strat_logger_->after_feature_calc_time = after_feature_calc_time;
    if (dry_run_) {
      LOG(INFO) << fmt::format(
          "product: {}, sell: {}, buy: {}",
          trade_product_->absolute_norm(), askp, bidp);
    } else {
      executor_->ManageMmOrders(
        curr_ts,
        askp,
        bidp,
        0,
        0,
        false,
        false);
    }
    if (enable_telemetry_) {
      auto acc_request = GetAccRequest(upd.mea());
      auto feed_latency_tag = GetFeedLatencyTag(upd);
      coin2::strategy::util::SetFeedTimeToLatency(
          strat_logger_,
          feed_mgr_,
          upd,
          feed_latency_tag,
          acc_request,
          true);
      coin2::strategy::util::SetFeedTimeToLatency(
          strat_logger_,
          feed_mgr_,
          upd,
          coin::proto::LatencyProto::ORDER_LATENCY,
          acc_request,
          false);
      // write health report
      auto strat_req = strat_logger_->GetStrategyRequestProto();
      auto symbol = upd.product().absolute_norm();
      //coin2::strategy::util::WriteFeedHealthReport(
      //    upd, strat_req, acc_request, symbol, health_reporter_);
    }
  }
}

}  // namespace coin2::strategy::dmm_strat
